package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.junit.{Before, Test}

case class Student(id: String, name: String, age: Int, gender: String, clazz: String)

case class Score(id: String, subject: String, score: Int)

case class Subject(subject: String, subjectName: String, fullScore: Int)

class Demo25StuExam {

  // 定义全局变量 让每一个方法中都能使用
  var sc: SparkContext = _

  var stuRDD: RDD[Student] = _
  var scoreRDD: RDD[Score] = _
  var subjectRDD: RDD[Subject] = _

  var sumScoreRDD: RDD[(String, Int)] = _

  var stuMap: Map[String, String] = _
  var subMap: Map[String, String] = _

  // 构建科目总分Map，以科目id作为key，科目总分作为Value
  var subFullScoMap: Map[String, Int] = _

  def initSparkContext(): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo25StuExam")
    conf.setMaster("local")

    sc = new SparkContext(conf)
  }

  def filterAndPrintWithIds(ids: Set[String]): Unit = {
    val idsBro: Broadcast[Set[String]] = sc.broadcast(ids)
    val stuMapBro: Broadcast[Map[String, String]] = sc.broadcast(stuMap)
    val subMapBro: Broadcast[Map[String, String]] = sc.broadcast(subMap)

    // 以分数数据scoList作为中间表（主表）进行遍历
    scoreRDD
      // 基于学生的id集合进行过滤
      .filter(sco => idsBro.value.contains(sco.id))
      // 逐条处理数据，借助Map 通过Key获取Value这种方式进行关联操作
      .map(sco => {
        // 通过id 从stuMap中获取学生的姓名和班级
        val nameAndClazz: String = stuMapBro.value.getOrElse(sco.id, "")
        // 通过subject科目id 从subMap中获取科目的名称
        val subjectName: String = subMapBro.value.getOrElse(sco.subject, "")
        s"${sco.id},$nameAndClazz,$subjectName,${sco.score}"
      })
      .foreach(println)
  }

  @Before
  // 该方法会在@Test标记的方法运行之前运行
  def init(): Unit = {
    initSparkContext()

    // 加载三类数据并将每一条数据转换成对应的样例类对象的格式 再放入List中
    // 使用JUNIT时默认的工作目录为当前模块的目录，所以这里不需要加载scala/前缀

    stuRDD = sc.textFile("data/students.txt").map(line => {
      val splits: Array[String] = line.split(",")
      Student(splits(0), splits(1), splits(2).toInt, splits(3), splits(4))
    })

    scoreRDD = sc.textFile("data/score.txt").map(line => {
      val splits: Array[String] = line.split(",")
      Score(splits(0), splits(1), splits(2).toInt)
    })

    subjectRDD = sc.textFile("data/subject.txt").map(line => {
      val splits: Array[String] = line.split(",")
      Subject(splits(0), splits(1), splits(2).toInt)
    })

    sumScoreRDD = scoreRDD
      // 取出id和分数
      .map(sco => {
        (sco.id, sco.score)
      })
      // 按照id进行分组
      .reduceByKey(_ + _)

    // 以id作为Key，name拼接clazz作为Value 构建stuMap
    stuMap = stuRDD
      .map(stu => {
        (stu.id, s"${stu.name},${stu.clazz}")
      }).collect().toMap

    // 以科目id subject 作为Key，subjectName作为Value 构建subMap
    subMap = subjectRDD
      .map(sub => {
        (sub.subject, sub.subjectName)
      }).collect().toMap

    // 构建科目总分Map，以科目id作为key，科目总分作为Value
    subFullScoMap = subjectRDD.map(sub => (sub.subject, sub.fullScore)).collect().toMap

  }

  @Test
  def printAll(): Unit = {
    stuRDD.take(10).foreach(println)
    scoreRDD.take(10).foreach(println)
    subjectRDD.take(10).foreach(println)
  }

  @Test
  // 3、统计每科都及格的学生 [学号，姓名，班级，科目，分数]
  def question03(): Unit = {
    val subFullScoMapBro: Broadcast[Map[String, Int]] = sc.broadcast(subFullScoMap)

    // 关联分数以及科目数据
    val ids: Set[String] = scoreRDD
      // 将每个学生及格的科目过滤出来
      .filter(sco => {
        val fullScore: Int = subFullScoMapBro.value.getOrElse(sco.subject, 0)
        sco.score >= fullScore * 0.6
      })
      // 统计每个学生及格的科目数
      .map(sco => (sco.id, 1))
      .reduceByKey(_ + _)
      // 过滤出及格的科目数为6 即表示所有科目都及格
      .filter(_._2 == 6)
      // 提取id
      .keys
      .collect()
      .toSet

    filterAndPrintWithIds(ids)

  }

  @Test
  // 4、统计每个班级的前三名 [学号，姓名，班级，分数]
  def question04(): Unit = {
    val stuMapBro: Broadcast[Map[String, String]] = sc.broadcast(stuMap)

    sumScoreRDD // 学生总分
      // 关联 stuMap 通过id获取学生的name以及clazz
      .map(kv => {
        val id: String = kv._1
        val nameAndClazz: String = stuMapBro.value.getOrElse(id, "")
        val splits: Array[String] = nameAndClazz.split(",")
        (id, splits(0), splits(1), kv._2)
      })
      // 将同一个班的数据进行分组处理
      .groupBy(_._3)
      .flatMap(kv => {
        // 按照总分降序排序 取前三名
        kv._2.toList.sortBy(-_._4).take(3).map(t4 => s"${t4._1},${t4._2},${t4._3},${t4._4}")
      }).foreach(println)
  }

  @Test
  // 5、统计偏科最严重的前100名学生  [学号，姓名，班级，科目，分数]
  def question05(): Unit = {

    val subFullScoMapBro: Broadcast[Map[String, Int]] = sc.broadcast(subFullScoMap)

    val ids: Set[String] = scoreRDD
      // 因为科目的总分不一致 所需需要先通过科目id关联获取科目总分，将所有成绩转换成百分制再进行处理
      .map(sco => {
        val fullScore: Int = subFullScoMapBro.value.getOrElse(sco.subject, 100)
        (sco.id, sco.score * 100.toDouble / fullScore)
      })
      // 按照id分组，将每个学生的6门科目成绩放在一组进行处理
      .groupBy(_._1)
      // 分别计算每个学生的平均分以及方差
      .map(kv => {
        val id: String = kv._1
        // 计算平均分
        val avgScore: Double = kv._2.map(_._2).sum / kv._2.size
        // 计算方差
        val variance: Double = kv._2.map(t2 => Math.pow(t2._2 - avgScore, 2)).sum / kv._2.size
        (id, variance)
      })
      .sortBy(_._2,ascending = false)
      .take(100)
      .map(_._1)
      .toSet

    filterAndPrintWithIds(ids)

  }


}
